package com.qdu.topn;

import java.io.IOException;
import java.util.Comparator;
import java.util.TreeMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopnReducer extends Reducer<Text, Text, Text, IntWritable> {
    private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>(new Comparator<Integer>() {
        public int compare(Integer a, Integer b) {
            return b-a;
        }
    });
    @Override
    public void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {

        for(Text value :values) {
            String[] nums = value.toString().split(",");
            repToRecordMap.put(Integer.parseInt(nums[8]),value.toString());
            if(repToRecordMap.size()>10) {
                repToRecordMap.remove(repToRecordMap.firstKey());
            }
        }

        for(String i:repToRecordMap.values()) {
            String[] result=i.split(",");
            context.write(new Text(result[1]), new IntWritable(Integer.parseInt(result[8])));
        }
    }

}